use std::{collections::HashMap,sync::{Arc}};
use once_cell::sync::Lazy;
use tokio::{io::AsyncWriteExt, net::TcpStream, sync::Mutex};
use crate::mq::protocol::Protocol;
use rusqlite::{params, Connection, OptionalExtension, Result};

const PACKAGE_END_FALG:u8 = 0x01;
pub struct Client {
    id: u64, //客户端ID，具有唯一性
    name: String, //客户端名称
    addr: String, //客户端绑定的TCP连接地址
}

impl Client {
    pub fn new(id: u64, name: String, addr: String) -> Self {
        Client {
            id,
            name,
            addr,
        }
    }
}

pub struct ClientMng{
    lists:HashMap<u64,Client>,
}
impl ClientMng {
    pub fn new()->Self{
        ClientMng { lists: HashMap::new() }
    }
    pub fn insert(&mut self,id:u64,name:String,addr:String){
        let client = Client::new(id.clone(), name.clone(), addr.clone());
        self.lists.insert(id, client);
    }
    pub fn remove(&mut self,id:u64){
        self.lists.remove(&id);
    }
    pub fn get_client(&mut self,id:u64)->Option<&mut Client>{
        self.lists.get_mut(&id)
    }
}
pub enum QoS{
    AtMostOnce=0,//至多一次
    AtLeastOnce=1,//至少一次
    ExactlyOnce=2,//仅一次
}
pub struct Topic{
    name:String,//主题名称,具有唯一值
    qos:QoS,//服务质量
    lists:HashMap<u64,u64>,//主题包含客户端ID列表
}
impl Topic {
    pub fn new(name:String)->Self{
        Topic { name,lists:HashMap::new(),qos:QoS::AtMostOnce }
    }
    pub fn set_qos(&mut self,qos:QoS)->&mut Topic{
        self.qos = qos;
        self
    }
    pub fn insert(&mut self,id:u64){
        self.lists.insert(id.clone(),id.clone());
    }
    pub fn remove(&mut self,id:u64){
        self.lists.remove(&id);
    }
    pub fn is_exist(&mut self,id:u64)->bool{
        self.lists.contains_key(&id)
    }
}

pub struct TopicMng{
    lists:HashMap<String,Topic>,//主题列表
}
impl TopicMng {
    pub fn new()->Self{
        TopicMng { lists: HashMap::new() }
    }
    pub fn insert(&mut self,name:String,id:u64){
        match self.lists.get_mut(&name) {
            Some(x)=>{
                x.insert(id.clone());
            }
            None=>{
                let topic = Topic::new(name.clone());
                self.lists.insert(name.clone(), topic);
                self.lists.get_mut(&name).unwrap().insert(id.clone());
            }
        }
    }
    pub fn get_topic(&mut self,name:String)->Option<&mut Topic>{
        self.lists.get_mut(&name)
    }
    pub fn remove(&mut self,name:String,id:u64){
        match self.lists.get_mut(&name) {
            Some(x)=>{
                x.remove(id.clone());
            }
            None=>{
            }
        }
    }
}

#[derive(Debug)]
struct KeyValue {
    key: String,
    value: String,
}

// #[derive(Debug,Clone)]
pub struct Broker {
    clients: HashMap<String, Arc<Mutex<TcpStream>>>, //客户端列表,key为addr
    clientmng:ClientMng,
    topicmng:TopicMng,
    db:Result<Connection>,
}

impl Broker{
    pub fn new()->Self{
        Broker {
        clients: HashMap::new(),
        clientmng:ClientMng::new(),
        topicmng:TopicMng::new(),
        db:Connection::open("mq.db"),
        }
    }
    
     fn get_keyvalue(&mut self,key: String) -> Result<Option<KeyValue>> {
        let conn = self.db.as_mut().unwrap();
        let mut stmt = conn.prepare("SELECT Key, Value FROM KeyValue WHERE Key = ?1")?;
        let keyvalue = stmt.query_row(params![key], |row| {
            Ok(KeyValue {
                key: row.get(0)?,
                value: row.get(1)?,
            })
        }).optional()?;
        
        Ok(keyvalue)            
    }
    pub fn get_value_from_db(&mut self,key:&str)->String{
        let aa = self.get_keyvalue(key.to_string()).unwrap();
        match aa {
            Some(s)=>{
                return s.value;
            }
            None=>{
                return "".to_string();
            }
        }
    }
    pub fn start(&mut self){
        let aa =self.get_value_from_db("Version");
        println!("Service Version={}",aa);
    }

    pub fn add_client(&mut self, client:TcpStream)->Option<&mut Arc<Mutex<TcpStream>>>{
        let addr = client.peer_addr().unwrap().to_string();      
        self.clients.insert(addr.clone(), Arc::new(Mutex::new(client)));
        println!("Connect Addr={:?},Count={}",addr,self.clients.len());
        self.clients.get_mut(&addr)
    }

    pub async fn remove_client(&mut self,addr:&String){
        self.clients.remove(addr);
        println!("DisConnect Addr={:?}",addr);
    }    

    pub async fn process_client_data(&mut self,addr:&String,data:Vec<u8>){
        println!("Data={:?},Addr={:?}",data,addr);
        let s = Protocol::new();
        s.Analyze(addr, &data);
    }

    pub fn insert_clientmng(&mut self,id:u64,name:String,addr:String){
        self.clientmng.insert(id.clone(), name.clone(), addr.clone());
    }
    pub fn remove_clientmng(&mut self,id:u64){
        self.clientmng.remove(id.clone());
    }

    pub async fn write_raw_data(&mut self,addr:&String, data: Vec<u8>)->bool {
        let stream = self.clients.get_mut(addr);
        if stream.is_none() {
            return false;
        }
        if let Err(e) = stream.unwrap().lock().await.write_all(&data).await {
            println!("Error writing to server: {:?}", e);
            return false;
        }
        return true;
    }

    pub async fn write_package_data(&mut self,addr:&String, data: Vec<u8>)->bool {
        let stream = self.clients.get_mut(addr);
        if stream.is_none() {
            return false;
        }
        let len = (data.len()+1) as u32;
        let mut packet = len.to_be_bytes().to_vec();
        packet.extend_from_slice(&data);
        packet.push(PACKAGE_END_FALG);
        if let Err(e) = stream.unwrap().lock().await.write_all(&packet).await {
            println!("Error writing to server: {:?}", e);
            return false;
        }
        return true;
    }
}

static BROKER_INST: Lazy<Arc<Mutex<Broker>>> = Lazy::new(|| {
    Arc::new(Mutex::new(Broker::new()))
});

// 获取单例的方法
pub fn get_broker() -> Arc<Mutex<Broker>> {
    // 克隆 Arc，增加引用计数
    BROKER_INST.clone()
}